#include "stream/impl/rocksdb/Topic.h"
#include <iostream>

#include "stream/impl/rocksdb/Meta.h"
#include "stream/impl/rocksdb/Data.h"
#include "stream/impl/rocksdb/meta/Size.h"

namespace stream {
namespace impl {
namespace rocksdb {

static std::string path(const std::string & dir, const std::string & name){
	return dir + "/" + name;
}
static std::string dataPath(const std::string & path){
	return path + ".meta" + ".db";
}
static std::string metaPath(const std::string & path){
	return path + ".data" + ".db";
}

Topic::Topic(const std::string & name, const std::string & dir)
: ::stream::Topic(name, dir)
, _path(path(dir, name))
, _meta(::stream::impl::rocksdb::makeMeta(metaPath(_path)))
, _data(::stream::impl::rocksdb::makeData(dataPath(_path)))
{
}
Topic::~Topic()
{
	this->close();
}

bool Topic::_open() { 
	std::lock_guard<std::mutex> lock(_mutex);

	bool ret = _meta->open();
	if(!ret){
		std::cerr << "open meta db failed" << std::endl;
		return false;
	}
	ret = _data->open();
	if(!ret) {
		std::cerr << "open data db failed" << std::endl;
		return false;
	}

	//restart logical 
	size_t sizeInMeta =_meta->size().load();
	size_t sizeInData = 0;
	if(false == _data->size(sizeInData)){
		std::cerr << "ERROR: get data size failed" << std::endl;
		return false;
	}

	if(sizeInMeta != sizeInData){
		_meta->size().set(sizeInData);
		ret = _meta->size().store();
		if(!ret){
			std::cerr << "ERROR: save size of Data in meta db failed" << std::endl;
			return false;
		}
	}
	return true;
}
void Topic::_close() {
	std::lock_guard<std::mutex> lock(_mutex);
	_data->close();

	bool ret = _meta->size().store();
	if(!ret){
		std::cerr << "ERROR: save size of Data in meta db failed" << std::endl;
	}
	_meta->close();
	std::cout << "topic real closed" << std::endl;
}

int Topic::append(const std::string & key, const std::string & value)  {
	std::lock_guard<std::mutex> lock(_mutex);

	int ret = _data->append(key, value);
	if(ret >= 0){
		_meta->size().inc();
		_meta->size().store();
	}
	return ret;
}

bool Topic::get(const std::string & key, std::string & value) {
	std::lock_guard<std::mutex> lock(_mutex);

	return _data->get(key, value);
}
size_t Topic::size() const {
	std::lock_guard<std::mutex> lock(_mutex);

	return _meta->size().get();
}
void Topic::clear() {
	std::lock_guard<std::mutex> lock(_mutex);

	_data->clear();

	_meta->size().set(0);
	_meta->size().store();
}

}//end of namespace rocksdb
}//end of namespace impl
}//end of namespace stream
